package com.bujunjie.study.realtime.dws.app;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.bujunjie.study.realtime.common.base.BaseApp;
import com.bujunjie.study.realtime.common.constant.FlinkConstant;
import com.bujunjie.study.realtime.common.function.Object2JsonStringFunction;
import com.bujunjie.study.realtime.common.model.sink.PageInfoModel;
import com.bujunjie.study.realtime.common.util.FlinkSinkUtil;
import com.bujunjie.study.realtime.dws.app.sink.UserLoginModel;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

/**
 * <p>用户登录</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/22 15:12
 */
public class DwsUserUserLoginWindow extends BaseApp {

    public static void main(String[] args) throws Exception {
        new DwsUserUserLoginWindow().start(
                10024,
                4,
                "dws_user_user_login_window",
                FlinkConstant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        KeyedStream<PageInfoModel, String> midKeyDS = kafkaStrDS.map(x -> JSONObject.parseObject(x, PageInfoModel.class))
                .keyBy(x -> x.getCommon().getMid());
        SingleOutputStreamOperator<PageInfoModel> filterDS = kafkaStrDS.process(new ProcessFunction<String, PageInfoModel>() {
            @Override
            public void processElement(String value, ProcessFunction<String, PageInfoModel>.Context ctx, Collector<PageInfoModel> out) throws Exception {
                PageInfoModel pageInfoModel = JSONObject.parseObject(value, PageInfoModel.class);
                String uid = pageInfoModel.getCommon().getUid();
                String lastPageId = pageInfoModel.getPage().getLastPageId();
                if (StrUtil.isNotBlank(uid) && "login".equals(lastPageId) || StrUtil.isEmpty(lastPageId)) {
                    out.collect(pageInfoModel);
                }
            }
        });

        SingleOutputStreamOperator<PageInfoModel> waterMarkDS = filterDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<PageInfoModel>forBoundedOutOfOrderness(Duration.ofSeconds(2 * 60))
                .withTimestampAssigner(new SerializableTimestampAssigner<PageInfoModel>() {
                    @Override
                    public long extractTimestamp(PageInfoModel element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        KeyedStream<PageInfoModel, String> keyedDS = waterMarkDS.keyBy(x -> x.getCommon().getMid());

        SingleOutputStreamOperator<UserLoginModel> loginDS = keyedDS.process(new KeyedProcessFunction<String, PageInfoModel, UserLoginModel>() {

            private ValueState<Long> lastLoginDateState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<Long> stateDescriptor = new ValueStateDescriptor<>("lastLoginDateState", Long.class);
                lastLoginDateState = getRuntimeContext().getState(stateDescriptor);

            }

            @Override
            public void processElement(PageInfoModel value, KeyedProcessFunction<String, PageInfoModel, UserLoginModel>.Context ctx, Collector<UserLoginModel> out) throws Exception {
                Long lastDate = lastLoginDateState.value();
                Long ts = value.getTs();

                Long uuCt = 0L;
                Long backCt = 0L;
                if (ObjectUtil.isNotNull(lastDate)) {
                    //若状态中的末次登录日期不为 null，进一步判断。
                    if (!lastDate.equals(ts)) {
                        //如果末次登录日期不等于当天日期则独立用户数 uuCt 记为 1，并将状态中的末次登录日期更新为当日，进一步判断。
                        uuCt = 1L;
                        lastLoginDateState.update(ts);
                        //如果当天日期与末次登录日期之差大于等于8天则回流用户数backCt置为1。
                        Long day = (ts - lastDate) / 1000 / 60 / 60 / 24;
                        if (day >= 8) {
                            backCt = 1L;
                        }
                    }
                } else {
                    //如果状态中的末次登录日期为 null，将 uuCt 置为 1，backCt 置为 0，并将状态中的末次登录日期更新为当日。
                    uuCt = 1L;
                    lastLoginDateState.update(ts);
                }

                Long zero = Long.parseLong(FlinkConstant.ZERO);

                if (!zero.equals(uuCt) || !zero.equals(backCt)) {
                    UserLoginModel userLoginModel = UserLoginModel
                            .builder()
                            .stt(StrUtil.EMPTY)
                            .edt(StrUtil.EMPTY)
                            .curDate(StrUtil.EMPTY)
                            .backCt(backCt)
                            .uuCt(uuCt)
                            .build();
                    out.collect(userLoginModel);
                }

            }
        });

        AllWindowedStream<UserLoginModel, TimeWindow> windowDS = loginDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<UserLoginModel> reduceDS = windowDS.reduce(new ReduceFunction<UserLoginModel>() {
            @Override
            public UserLoginModel reduce(UserLoginModel value1, UserLoginModel value2) throws Exception {
                value1.setUuCt(value1.getUuCt() + value2.getUuCt());
                value1.setBackCt(value1.getBackCt() + value2.getBackCt());
                return value1;
            }
        }, new AllWindowFunction<UserLoginModel, UserLoginModel, TimeWindow>() {
            @Override
            public void apply(TimeWindow window, Iterable<UserLoginModel> values, Collector<UserLoginModel> out) throws Exception {
                UserLoginModel bean = values.iterator().next();
                String stt = DateUtil.format(new Date(window.getStart()), DatePattern.NORM_DATETIME_PATTERN);
                String edt = DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATETIME_PATTERN);
                String curDate = DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATE_PATTERN);
                bean.setStt(stt);
                bean.setEdt(edt);
                bean.setCurDate(curDate);
                out.collect(bean);
            }
        });

        // 8.将聚合结果写到Doris
        reduceDS.print();
        reduceDS
                .map(new Object2JsonStringFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_user_user_login_window"));
    }
}
